import tkinter as tk
from tkinter import scrolledtext, messagebox
from kafka import KafkaConsumer, KafkaProducer, KafkaAdminClient
import threading
import random
import string

# Kafka配置
KAFKA_BOOTSTRAP_SERVERS_DEFAULT = 'localhost:9092'


class KafkaConsumerApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Kafka Consumer GUI")
        self.root.geometry("1200x800")  # 设置窗口大小

        self.kafka_bootstrap_servers = tk.StringVar(value=KAFKA_BOOTSTRAP_SERVERS_DEFAULT)

        # 左上角：Kafka地址和连接按钮
        self.left_top_frame = tk.Frame(root)
        self.left_top_frame.pack(side=tk.TOP, fill=tk.X)

        self.label = tk.Label(self.left_top_frame, text="Kafka Bootstrap Servers:")
        self.label.pack(side=tk.LEFT, padx=5, pady=5)

        self.entry = tk.Entry(self.left_top_frame, textvariable=self.kafka_bootstrap_servers, width=30)
        self.entry.pack(side=tk.LEFT, padx=5, pady=5)

        self.connect_button = tk.Button(self.left_top_frame, text="Connect", command=self.connect_kafka)
        self.connect_button.pack(side=tk.LEFT, padx=5, pady=5)

        self.disconnect_button = tk.Button(self.left_top_frame, text="Disconnect", command=self.disconnect_kafka, state=tk.DISABLED)
        self.disconnect_button.pack(side=tk.LEFT, padx=5, pady=5)

        # 右上角：主题列表
        self.right_top_frame = tk.Frame(root)
        self.right_top_frame.pack(side=tk.TOP, fill=tk.X)

        self.topic_listbox = tk.Listbox(self.right_top_frame, width=50, height=10)
        self.topic_listbox.pack(side=tk.LEFT, padx=10, pady=10)
        self.topic_listbox.bind('<<ListboxSelect>>', self.on_topic_select)

        # 左下角：接收消息
        self.left_bottom_frame = tk.Frame(root)
        self.left_bottom_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

        self.message_text = scrolledtext.ScrolledText(self.left_bottom_frame, wrap=tk.WORD, width=40, height=20)
        self.message_text.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)

        self.start_button = tk.Button(self.left_bottom_frame, text="Start Consuming", command=self.start_consuming, state=tk.DISABLED)
        self.start_button.pack(side=tk.TOP, padx=5, pady=5)

        self.stop_button = tk.Button(self.left_bottom_frame, text="Stop Consuming", command=self.safe_stop_consuming, state=tk.DISABLED)
        self.stop_button.pack(side=tk.TOP, padx=5, pady=5)

        self.clear_button = tk.Button(self.left_bottom_frame, text="Clear Messages", command=self.clear_messages)
        self.clear_button.pack(side=tk.TOP, padx=5, pady=5)

        # 右下角：发送消息
        self.right_bottom_frame = tk.Frame(root)
        self.right_bottom_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)

        self.send_message_label = tk.Label(self.right_bottom_frame, text="Message to Send:")
        self.send_message_label.pack(side=tk.TOP, padx=5, pady=5)

        self.send_message_entry = tk.Entry(self.right_bottom_frame, width=40)
        self.send_message_entry.pack(side=tk.TOP, padx=5, pady=5)

        self.send_button = tk.Button(self.right_bottom_frame, text="Send Message", command=self.send_message, state=tk.DISABLED)
        self.send_button.pack(side=tk.TOP, padx=5, pady=5)

        self.consumer = None
        self.producer = None
        self.consuming = False
        self.consumer_thread = None
        self.selected_topic = None
        self.connected = False

    def connect_kafka(self):
        bootstrap_servers = self.kafka_bootstrap_servers.get()
        try:
            admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
            topics = admin_client.list_topics()
            self.topic_listbox.delete(0, tk.END)
            for topic in topics:
                self.topic_listbox.insert(tk.END, topic)
            messagebox.showinfo("Success", "Connected to Kafka successfully!")
            self.connected = True
            self.connect_button.config(state=tk.DISABLED)
            self.disconnect_button.config(state=tk.NORMAL)
            self.send_button.config(state=tk.NORMAL)
        except Exception as e:
            messagebox.showerror("Error", f"Failed to connect to Kafka: {e}")

    def disconnect_kafka(self):
        if self.connected:
            self.stop_consuming()
            self.topic_listbox.delete(0, tk.END)
            #self.kafka_bootstrap_servers.set(KAFKA_BOOTSTRAP_SERVERS_DEFAULT)
            messagebox.showinfo("Success", "Disconnected from Kafka successfully!")
            self.connected = False
            self.connect_button.config(state=tk.NORMAL)
            self.disconnect_button.config(state=tk.DISABLED)
            self.start_button.config(state=tk.DISABLED)
            self.stop_button.config(state=tk.DISABLED)
            self.send_button.config(state=tk.DISABLED)

    def on_topic_select(self, event):
        selected_indices = self.topic_listbox.curselection()
        if selected_indices:
            self.selected_topic = self.topic_listbox.get(selected_indices[0])
            self.start_button.config(state=tk.NORMAL)

    def start_consuming(self):
        if not self.consuming and self.selected_topic:
            self.consuming = True
            group_id = ''.join(random.choices(string.ascii_letters + string.digits, k=10))
            self.consumer = KafkaConsumer(
                self.selected_topic,
                bootstrap_servers=self.kafka_bootstrap_servers.get(),
                auto_offset_reset='latest',  # 从最新的消息开始
                group_id=group_id
            )
            self.consumer_thread = threading.Thread(target=self.consume_messages)
            self.consumer_thread.start()
            self.start_button.config(state=tk.DISABLED)
            self.stop_button.config(state=tk.NORMAL)

    def stop_consuming(self):
        if self.consuming:
            self.consuming = False
            self.consumer_thread.join()  # 等待消费者线程结束
            self.consumer.close()

    def safe_stop_consuming(self):
        if self.consuming:
            self.consuming = False
            self.stop_button.config(state=tk.DISABLED)
            self.start_button.config(state=tk.NORMAL)
            # 启动一个单独的线程来处理关闭操作
            threading.Thread(target=self.stop_consuming).start()

    def consume_messages(self):
        try:
            while self.consuming:
                msg = next(self.consumer)
                if msg is not None:
                    self.message_text.insert(tk.END, f"Received message: {msg.value.decode('utf-8')}\n")
                    #self.message_text.insert(tk.END, f"Received message: {len(msg.headers)}:{len(msg.value)}\n")
                    self.message_text.see(tk.END)
        except Exception as e:
            print(f"Error consuming messages: {e}")
        finally:
            self.consumer.close()

    def send_message(self):
        if self.selected_topic:
            message = self.send_message_entry.get()
            if message:
                if not self.producer:
                    self.producer = KafkaProducer(bootstrap_servers=self.kafka_bootstrap_servers.get())
                try:
                    self.producer.send(self.selected_topic, message.encode('utf-8'))
                    self.producer.flush()
                    messagebox.showinfo("Success", "Message sent successfully!")
                except Exception as e:
                    messagebox.showerror("Error", f"Failed to send message: {e}")
            else:
                messagebox.showwarning("Warning", "Message cannot be empty!")
        else:
            messagebox.showwarning("Warning", "Please select a topic first!")

    def clear_messages(self):
        self.message_text.delete(1.0, tk.END)


if __name__ == "__main__":
    root = tk.Tk()
    app = KafkaConsumerApp(root)
    root.mainloop()
